package com.atguigu.edu.app.dws;

import bean.TrafficPageViewBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;
import util.DateFormatUtil;
import util.MyClickhouseUtil;
import util.MyKafkaUtil;

import java.time.Duration;

/**
 * 流量域：版本、地区、新老访客 维度聚合统计
 */
public class DwsTrafficVcChArIsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.检查点相关设置
       /* env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, org.apache.flink.api.common.time.Time.days(30), org.apache.flink.api.common.time.Time.seconds(3)));
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://39.101.79.248:8020//edu_realtime/checkpoint");
        System.setProperty("HADOOP_USER_NAME", "atguigu");*/
        //TODO 3.分别从kafka的页面浏览、独立访客、用户跳出主题读取数据，并转化为流
        String pageLogTopic = "dwd_traffic_page_log";
        String uvTopic = "dwd_traffic_unique_visitor_detail";
        String ujpTopic = "dwd_traffic_user_jump_detail";
        String groupId = "dws_vc_ch_ar_isnew_group";
        DataStreamSource<String> pageLogStrDS = env.addSource(MyKafkaUtil.getKafkaConsumer(pageLogTopic, groupId));
        DataStreamSource<String> uvStrDS = env.addSource(MyKafkaUtil.getKafkaConsumer(uvTopic, groupId));
        DataStreamSource<String> ujpStrDS = env.addSource(MyKafkaUtil.getKafkaConsumer(ujpTopic, groupId));
        //TODO 4.数据类型转换 jsonStr->实体类

        //页面
        SingleOutputStreamOperator<TrafficPageViewBean> pageVisitDS = pageLogStrDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String jsonStr) throws Exception {
                //先将jsonStr转为jsonObj
                JSONObject pageJson = JSON.parseObject(jsonStr);
                JSONObject commonJson = pageJson.getJSONObject("common");
                //再new对象进行赋值
                TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean("", "", commonJson.getString("vc"), commonJson.getString("ch"), commonJson.getString("ar"), commonJson.getString("is_new"), 0L, 0L, 1L, pageJson.getJSONObject("page").getLong("during_time"), 0L, pageJson.getLong("ts"));
                //如果是上页id为空，则算作一个独立访客数
                String last_page_id = pageJson.getJSONObject("page").getString("last_page_id");
                if (StringUtils.isEmpty(last_page_id)) {
                    trafficPageViewBean.setSvCt(1L);
                }
                return trafficPageViewBean;
            }
        });


        //独立访客
        SingleOutputStreamOperator<TrafficPageViewBean> uvDS = uvStrDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String jsonStr) throws Exception {
                JSONObject uvJson = JSON.parseObject(jsonStr);
                JSONObject commonJson = uvJson.getJSONObject("common");
                TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean("", "", commonJson.getString("vc"), commonJson.getString("ch"), commonJson.getString("ar"), commonJson.getString("is_new"), 1L, 0L, 0L, 0L, 0L, uvJson.getLong("ts"));
                return trafficPageViewBean;
            }
        });

        //跳出
        SingleOutputStreamOperator<TrafficPageViewBean> ujpDS = ujpStrDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String jsonStr) throws Exception {
                JSONObject ujpJson = JSON.parseObject(jsonStr);
                JSONObject commonJson = ujpJson.getJSONObject("common");
                TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean("", "", commonJson.getString("vc"), commonJson.getString("ch"), commonJson.getString("ar"), commonJson.getString("is_new"), 0L, 0L, 0L, 0L, 1L, ujpJson.getLong("ts"));
                return trafficPageViewBean;
            }
        });

        //TODO 5.将3条流进行union
        DataStream<TrafficPageViewBean> unionDS = pageVisitDS.union(uvDS, ujpDS);

        //TODO 6.指定时间戳和水位线
        SingleOutputStreamOperator<TrafficPageViewBean> withWatermarkDS = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
            @Override
            public long extractTimestamp(TrafficPageViewBean pageViewBean, long recordTimestamp) {
                return pageViewBean.getTs();
            }
        }));

        //TODO 7.按照维度进行分组
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyedDS = withWatermarkDS.keyBy(
                new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {

            @Override
            public Tuple4<String, String, String, String> getKey(TrafficPageViewBean pageViewBean) throws Exception {
                return Tuple4.of(pageViewBean.getVc(), pageViewBean.getCh(), pageViewBean.getAr(), pageViewBean.getIsNew());

            }
        });
        //TODO 8.开窗
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        //TODO 9.聚合计算
        SingleOutputStreamOperator<TrafficPageViewBean> reduceDS = windowDS.reduce(new ReduceFunction<TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                value1.setUjCt(value1.getUjCt() + value2.getUjCt());
                return value1;
            }
        }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple4<String, String, String, String> tuple4, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                for (TrafficPageViewBean pageViewBean : input) {
                    pageViewBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                    pageViewBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                    pageViewBean.setTs(System.currentTimeMillis());
                    //向下游传递流
                    out.collect(pageViewBean);
                }
            }
        });
        reduceDS.print(">>>>>");
        //TODO 10
        // 将聚合的结果写到clickhouse
reduceDS.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_traffic_vc_ch_ar_is_new_page_view_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));
        env.execute();
    }
}